#ifndef __M_MSG_H__
#define __M_MSG_H__
#include "../mqcommon/mq_logger.hpp" // 日志工具类，提供DLOG等日志输出宏
#include "../mqcommon/mq_helper.hpp" // 文件操作辅助类，提供文件创建、读写等功能
#include "../mqcommon/mq_msg.pb.h"   // Protobuf消息定义，包含Message等消息结构
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <list>

namespace zdfmq
{
// 数据文件后缀名定义
#define DATAFILE_SUBFIX ".mqd"    // 主数据文件后缀
#define TMPFILE_SUBFIX ".mqd.tmp" // 临时文件后缀（用于垃圾回收）

    // 消息智能指针类型定义，简化shared_ptr<zdfmq::Message>的使用
    using MessagePtr = std::shared_ptr<zdfmq::Message>;

    /**
     * @brief 消息映射器，负责消息的持久化存储（文件操作）
     * 管理消息在磁盘文件中的读写、创建、删除等操作，实现消息的持久化
     */
    class MessageMapper
    {
    public:
        /**
         * @brief 构造函数，初始化消息文件路径
         * @param basedir 基础目录路径
         * @param qname 队列名称（用于生成文件名）
         */
        MessageMapper(std::string &basedir, const std::string &qname) : _qname(qname)
        {
            // 确保基础目录路径以'/'结尾
            if (basedir.back() != '/')
                basedir.push_back('/');
            // 生成数据文件和临时文件的完整路径
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
            // 检查基础目录是否存在，不存在则创建
            if (FileHelper(basedir).exists() == false)
            {
                assert(FileHelper::createDirectory(basedir)); // 断言确保目录创建成功
            }
            // 创建消息数据文件（如果不存在）
            createMsgFile();
        }

        /**
         * @brief 创建消息数据文件（如果不存在）
         * @return 成功返回true，失败返回false
         */
        bool createMsgFile()
        {
            // 如果文件已存在，直接返回成功
            if (FileHelper(_datafile).exists() == true)
            {
                return true;
            }
            // 创建新文件
            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                DLOG("创建队列数据文件 %s 失败！", _datafile.c_str());
                return false;
            }
            return true;
        }

        /**
         * @brief 删除消息相关文件（主数据文件和临时文件）
         */
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }

        /**
         * @brief 插入消息到主数据文件
         * @param msg 待插入的消息指针
         * @return 插入成功返回true，失败返回false
         */
        bool insert(MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }

        /**
         * @brief 从文件中删除消息（逻辑删除，标记为无效）
         * @param msg 待删除的消息指针
         * @return 操作成功返回true，失败返回false
         */
        bool remove(MessagePtr &msg)
        {
            // 1. 将消息的有效标志位设为'0'（逻辑删除）
            msg->mutable_payload()->set_valid("0");
            // 2. 序列化消息体
            std::string body = msg->payload().SerializeAsString();
            // 检查序列化后的数据长度是否与原长度一致（确保覆盖写入时不破坏文件结构）
            if (body.size() != msg->length())
            {
                DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致!");
                return false;
            }
            // 3. 覆盖写入到文件中消息的原始位置
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                DLOG("向队列数据文件写入数据失败！");
                return false;
            }
            return true;
        }

        /**
         * @brief 垃圾回收，清理无效消息并重建数据文件
         * @return 回收后剩余的有效消息列表
         */
        std::list<MessagePtr> gc()
        {
            bool ret;
            std::list<MessagePtr> result;
            // 1. 加载所有有效消息
            ret = load(result);
            if (ret == false)
            {
                DLOG("加载有效数据失败！\n");
                return result;
            }
            // 2. 创建临时文件并写入有效消息
            FileHelper::createFile(_tmpfile);
            for (auto &msg : result)
            {
                DLOG("向临时文件写入数据: %s", msg->payload().body().c_str());
                ret = insert(_tmpfile, msg);
                if (ret == false)
                {
                    DLOG("向临时文件写入消息数据失败！！");
                    return result;
                }
            }
            // 3. 删除原数据文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                DLOG("删除源文件失败！");
                return result;
            }
            // 4. 将临时文件重命名为原数据文件（完成替换）
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                DLOG("修改临时文件名称失败！");
                return result;
            }
            // 5. 返回有效消息列表
            return result;
        }

    private:
        /**
         * @brief 从数据文件加载所有有效消息
         * @param result 输出参数，存储加载到的有效消息
         * @return 加载成功返回true，失败返回false
         */
        bool load(std::list<MessagePtr> &result)
        {
            // 文件存储格式：[4字节长度][数据][4字节长度][数据]...
            FileHelper data_file_helper(_datafile);
            size_t offset = 0;                      // 当前读取偏移量
            size_t msg_size;                        // 消息体长度
            size_t fsize = data_file_helper.size(); // 文件总大小
            bool ret;

            // 循环读取所有消息
            while (offset < fsize)
            {
                // 读取消息长度（4字节）
                ret = data_file_helper.read((char *)&msg_size, offset, sizeof(size_t));
                if (ret == false)
                {
                    DLOG("读取消息长度失败！");
                    return false;
                }
                offset += sizeof(size_t); // 偏移量移动到消息体起始位置

                // 读取消息体数据
                std::string msg_body(msg_size, '\0');
                data_file_helper.read(&msg_body[0], offset, msg_size);
                if (ret == false)
                {
                    DLOG("读取消息数据失败！");
                    return false;
                }
                offset += msg_size; // 偏移量移动到下一个消息长度位置

                // 解析消息并检查有效性
                MessagePtr msgp = std::make_shared<Message>();
                msgp->mutable_payload()->ParseFromString(msg_body);
                // 跳过无效消息（逻辑删除的消息）
                if (msgp->payload().valid() == "0")
                {
                    DLOG("加载到无效消息：%s", msgp->payload().body().c_str());
                    continue;
                }
                // 有效消息加入结果列表
                result.push_back(msgp);
            }
            return true;
        }

        /**
         * @brief 向指定文件插入消息（私有重载，支持主文件和临时文件）
         * @param filename 目标文件路径
         * @param msg 待插入的消息指针
         * @return 插入成功返回true，失败返回false
         */
        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 新增消息追加到文件末尾
            // 1. 序列化消息体
            std::string body = msg->payload().SerializeAsString();
            // 2. 获取文件当前大小（作为写入起始位置）
            FileHelper helper(filename);
            size_t fsize = helper.size();
            size_t msg_size = body.size();

            // 写入逻辑：先写消息长度（4字节），再写消息体
            bool ret = helper.write((char *)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                DLOG("向队列数据文件写入数据长度失败！");
                return false;
            }
            // 写入消息体
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
            if (ret == false)
            {
                DLOG("向队列数据文件写入数据失败！");
                return false;
            }
            // 更新消息对象中的存储信息（偏移量和长度）
            msg->set_offset(fsize + sizeof(size_t)); // 消息体在文件中的起始位置
            msg->set_length(body.size());            // 消息体长度
            return true;
        }

    private:
        std::string _qname;    // 队列名称
        std::string _datafile; // 主数据文件路径
        std::string _tmpfile;  // 临时文件路径（用于垃圾回收）
    };

    /**
     * @brief 队列消息管理器，负责单个队列的消息内存管理和持久化协调
     * 维护待推送消息、待确认消息、持久化消息等状态，处理消息的入队、出队、确认等操作
     */
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>; // 智能指针类型定义

        /**
         * @brief 构造函数，初始化队列消息管理器
         * @param basedir 基础目录路径（用于MessageMapper）
         * @param qname 队列名称
         */
        QueueMessage(std::string &basedir, const std::string &qname) : _mapper(basedir, qname),
                                                                       _qname(qname), _valid_count(0), _total_count(0) {}

        /**
         * @brief 恢复队列消息（从持久化文件加载有效消息到内存）
         * @return 恢复成功返回true
         */
        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
            // 调用MessageMapper的gc()加载有效消息（同时完成文件清理）
            _msgs = _mapper.gc();
            // 将加载的消息加入持久化消息哈希表
            for (auto &msg : _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            // 更新计数（持久化消息总数=有效数）
            _valid_count = _total_count = _msgs.size();
            return true;
        }

        /**
         * 向队列插入消息，支持持久化存储
         *
         * @param bp 消息属性（包含持久化模式、ID等），可为nullptr
         * @param body 消息体内容
         * @param queue_is_durable 队列是否为持久化队列（影响消息是否持久化存储）
         * @return 插入成功返回true，失败返回false
         */
        bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 1. 构造消息对象并设置基本属性
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);

            // 2. 根据参数设置消息的持久化模式和其他属性
            if (bp != nullptr)
            {
                // 若提供了属性参数，则使用其中的持久化模式（但受队列持久化状态约束）
                // 队列持久化时使用属性中的模式，否则强制为非持久化
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                // 若未提供属性参数，则使用默认值
                // 队列持久化时默认使用持久化模式，否则使用非持久化模式
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            // 3. 加锁确保线程安全
            std::unique_lock<std::mutex> lock(_mutex);

            // 4. 判断消息是否需要持久化存储
            if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 标记消息为有效状态（用于持久化存储）
                msg->mutable_payload()->set_valid("1");

                // 持久化存储消息到文件系统
                bool ret = _mapper.insert(msg);
                if (ret == false)
                {
                    DLOG("持久化存储消息：%s 失败了！", body.c_str());
                    return false;
                }

                // 更新持久化消息统计信息
                _valid_count += 1; // 有效持久化消息数量+1
                _total_count += 1; // 总持久化消息数量+1

                // 将消息添加到持久化消息哈希表（ID -> 消息）
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }

            // 5. 将消息添加到内存队列（无论是否持久化都需要在内存中管理）
            _msgs.push_back(msg);
            return true;
        }

        /**
         * @brief 获取队首消息（出队操作）
         * @return 队首消息的智能指针，队列为空时返回空指针
         */
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
            if (_msgs.size() == 0)
            {
                return MessagePtr(); // 队列为空，返回空指针
            }
            // 取出队首消息
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 将消息加入待确认哈希表（等待消费者确认）
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }

        /**
         * @brief 移除消息（消费者确认后调用）
         * @param msg_id 消息ID
         * @return 操作成功返回true，失败返回false
         */
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
            // 1. 从待确认哈希表查找消息
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                DLOG("没有找到要删除的消息：%s!", msg_id.c_str());
                return true; // 消息不存在，视为成功
            }

            // 2. 处理持久化消息（更新文件标记）
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 调用MessageMapper标记消息为无效
                _mapper.remove(it->second);
                // 更新哈希表和计数
                _durable_msgs.erase(msg_id);
                _valid_count -= 1; // 有效持久化消息数-1
                gc();              // 检查是否需要垃圾回收
            }

            // 3. 从待确认哈希表移除
            _waitack_msgs.erase(msg_id);
            return true;
        }

        /**
         * @brief 获取待推送消息数量
         * @return 待推送消息数
         */
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        /**
         * @brief 获取持久化消息总数（包含无效消息）
         * @return 总持久化消息数
         */
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }

        /**
         * @brief 获取有效持久化消息数量
         * @return 有效持久化消息数
         */
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        /**
         * @brief 获取待确认消息数量
         * @return 待确认消息数
         */
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        /**
         * @brief 清空队列所有消息（内存和文件）
         */
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 删除持久化文件
            _mapper.removeMsgFile();
            // 清空所有内存容器
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            // 重置计数
            _valid_count = 0;
            _total_count = 0;
        }

    private:
        /**
         *@brief 检查是否需要执行垃圾回收
         *@return 需要回收返回 true，否则返回 false
         */
        bool GCCheck()
        {
            // 垃圾回收触发条件：
            // 1. 持久化消息总数超过 2000
            // 2. 有效消息占比低于 50%（通过整数运算避免浮点误差）
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
            {
                return true;
            }
            return false;
        }
        /**
         *@brief 执行垃圾回收（清理无效消息，优化文件存储）
         */
        void gc()
        {
            // 检查是否满足回收条件
            if (GCCheck() == false)
                return;
            // 调用 MessageMapper 的 gc () 清理文件并获取有效消息
            std::list<MessagePtr> msgs = _mapper.gc();
            // 更新内存中的持久化消息信息
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    // 异常处理：文件中存在内存未管理的消息，重新加入待推送队列
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理！");
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    continue;
                }
                // 更新消息在文件中的偏移量和长度（文件结构已改变）
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 更新计数（回收后总数 = 有效数）
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::mutex _mutex;                                         // 互斥锁，确保线程安全
        std::string _qname;                                        // 队列名称
        size_t _valid_count;                                       // 有效持久化消息数量
        size_t _total_count;                                       // 总持久化消息数量（含无效）
        MessageMapper _mapper;                                     // 消息持久化映射器
        std::list<MessagePtr> _msgs;                               // 待推送消息队列（内存）
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息哈希表（ID-> 消息）
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息哈希表（ID-> 消息）
    };

    /**
     *@brief 消息管理器，负责管理多个队列的消息操作
     *作为上层接口，提供队列的初始化、销毁、消息插入、获取、确认等功能
     */
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>; // 智能指针类型定义
        /**
         *@brief 构造函数，初始化基础目录
         *@param basedir 持久化文件的基础目录
         */
        MessageManager(const std::string &basedir) : _basedir(basedir) {}
        /**
        @brief 清空所有队列的消息（内存和文件）
        */
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
            for (auto &qmsg : _queue_msgs)
            {
                qmsg.second->clear();
            }
        }
        /**
         *@brief 初始化队列消息管理器（如果不存在）
         *@param qname 队列名称
         */
        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 检查队列是否已初始化
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                {
                    return;
                }
                // 创建新的队列消息管理器
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            // 恢复队列消息（从文件加载）
            qmp->recovery();
        }
        /**
         *@brief 销毁队列消息管理器（清理资源）
         *@param qname 队列名称
         */
        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 查找队列
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    return;
                }
                // 保存智能指针并从哈希表移除
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            // 清空队列消息（内存和文件）
            qmp->clear();
        }
        /**
        @brief 向队列插入消息
        @param qname 队列名称
        @param bp 消息属性
        @param body 消息体
        @param queue_is_durable 队列是否持久化
        @return 插入成功返回 true，失败返回 false
        */
       // 这个最后一个参数是bool类型的参数，而上面你传入的是枚举类型的参数，枚举值从1开始的，所以不管是durable还是undurable都是true，所以都会
       // 进入持久化的判断中，所以total_count都是5了
        bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 查找队列
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("向队列 % s 新增消息失败：没有找到消息管理句柄！", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            // 调用队列的插入方法
            return qmp->insert(bp, body, queue_is_durable);
        }
        /**
         *@brief 获取队列的队首消息
         *@param qname 队列名称
         *@return 消息智能指针，队列不存在或为空时返回空指针
         */
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 查找队列
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 % s 队首消息失败：没有找到消息管理句柄！", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            // 调用队列的 front () 方法
            return qmp->front();
        }
        /**
         *@brief 确认消息（消费者处理完成后调用）
         *@param qname 队列名称
         *@param msg_id 消息 ID
         */
        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 查找队列
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("确认队列 % s 消息 % s 失败：没有找到消息管理句柄！", qname.c_str(), msg_id.c_str());
                    return;
                }
                qmp = it->second;
            }
            // 调用队列的 remove () 方法（确认消息）
            qmp->remove(msg_id);
            return;
        }
        /**
         *@brief 获取队列的待推送消息数量
         *@param qname 队列名称
         *@return 待推送消息数，队列不存在返回 0
         */
        size_t getable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 查找队列
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 % s 待推送消息数量失败：没有找到消息管理句柄！", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }
        /**
         *@brief 获取队列的总持久化消息数量（含无效）
         *@param qname 队列名称
         *@return 总持久化消息数，队列不存在返回 0
         */
        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 查找队列
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 % s 总持久化消息数量失败：没有找到消息管理句柄！", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }
        /**
         *@brief 获取队列的有效持久化消息数量
         *@param qname 队列名称
         *@return 有效持久化消息数，队列不存在返回 0
         */
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 查找队列
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 % s 有效持久化消息数量失败：没有找到消息管理句柄！", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        /**
         *@brief 获取队列的待确认消息数量
         *@param qname 队列名称
         *@return 待确认消息数，队列不存在返回 0
         */
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex); // 加锁确保线程安全
                // 查找队列
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 % s 待确认消息数量失败：没有找到消息管理句柄！", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }

    private:
        std::mutex _mutex;                                              // 互斥锁，确保线程安全
        std::string _basedir;                                           // 持久化文件基础目录
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; // 队列消息管理器哈希表（队列名 -> 管理器）
    };
}
#endif